package com.iiixxxiii.pf.api.kafka;

import com.jfinal.plugin.IPlugin;
import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
/**
 * jFinal kafka插件
 *
 */
public class KafkaPlugin implements IPlugin {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    public KafkaPlugin(String name,
                       String servers,
                       String keySerializer,
                       String valueSerializer) {
        KafkaTools.addProducer(name, servers, keySerializer, valueSerializer);
    }
    @Override
    public boolean start() {
        return true;
    }
    @Override
    public boolean stop() {
        logger.info("销毁所有生产者和消费者开始");
        for (Map.Entry<String, Producer> entry : KafkaTools.producerMap.entrySet()) {
            entry.getValue().close();
        }
        for (Map.Entry<String, KafkaConsumerTemplate> entry : KafkaTools.consumerMap.entrySet()) {
            entry.getValue().setRunner(false);
        }
        logger.info("销毁所有生产者和消费者结束");
        return true;
    }
}